package cn.iocoder.yudao.module.measure.mq.rabbit;

import cn.iocoder.yudao.framework.common.util.spring.SpringUtils;
import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder;
import cn.iocoder.yudao.module.measure.dal.dataobject.device.message.DeviceMessage;
import cn.iocoder.yudao.module.measure.dal.dataobject.device.message.adapter.DeviceMessageAdapter;
import cn.iocoder.yudao.module.measure.dal.dataobject.protocol.TemplateDeviceUploadDataConfig;
import cn.iocoder.yudao.module.measure.dal.dataobject.protocol.TemplateDeviceUploadMessage;
import cn.iocoder.yudao.module.measure.dal.dataobject.protocol.TemplateDeviceUploadProtocol;
import cn.iocoder.yudao.module.measure.dal.dataobject.upload.BusinessDeviceUploadData;
import cn.iocoder.yudao.module.measure.dal.dataobject.warning.BusinessDeviceWarnRecord;
import cn.iocoder.yudao.module.measure.dal.redis.RedisKeyConstants;
import cn.iocoder.yudao.module.measure.events.BusinessDeviceUploadDataEvent;
import cn.iocoder.yudao.module.measure.events.BusinessDeviceWarnRecordEvent;
import cn.iocoder.yudao.module.measure.service.protocol.TemplateDeviceUploadProtocolService;
import cn.iocoder.yudao.module.measure.service.upload.BusinessDeviceUploadDataService;
import cn.iocoder.yudao.module.measure.service.warning.BusinessDeviceUploadWarningConfigService;
import com.alibaba.fastjson.JSONObject;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApi;
import com.influxdb.client.WriteOptions;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.lnhuachuan.iot.common.BitDataTypeEnum;
import com.rabbitmq.client.Channel;
import com.zhengbzh.common.utils.HexUtils;
import com.zhengbzh.common.utils.LocalDateTimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.ByteOrder;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Component
@Slf4j
public class Receiver {



    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Resource
    private BusinessDeviceUploadDataService businessDeviceUploadDataService;

    @Autowired
    private TemplateDeviceUploadProtocolService templateDeviceUploadProtocolService;


    @Autowired
    private RedisTemplate redisTemplate;

//    @Autowired
//    private BusinessDeviceUploadWarningConfigService warningConfigService;

    @Autowired
    private ApplicationContext applicationContext;
//



    /**
     * 消费mq队列中的消息
     * 注意 autoStartup需要设置为false 否则消息消费会在init-method阶段执行 导致处理消息时可能短暂的无法获取其他bean
     * 启动命令在RabbitApplicationRunner中 通过消费者的id进行启动
     * ackMode设置为false,在消息处理完之前不返回ack消息 保障消息能够被正确入库
     * concurrency 设置消费者数量  值越大一般处理能力越强 相当于开了多线程  但是线程过多反而会影响效率 修改这个值时候需要注意测试实际效果
     * queues 设置接收的队列名称
     *
     * 处理出错的消息  会被发送到rawDataErrorQueues队列进行存储
     *
     * 步骤1.接收到mq消息，进入此方法
     * 步骤2.取消息中监测点id 来获取协议信息 先去缓存中查找 缓存中查不到则去数据库查找 查找数据库时会忽略租户id的限制
     * 步骤3.取出协议中租户id 和 协议具体内容 对消息进行解析
     * 步骤4.数据存储mysql数据库
     * 步骤5.发布设备上传数据事件，由DeviceDataCheckListener ， DeviceWriteToInfluxDBListener 处理
     * 步骤6.发布设备上传的报警事件，由DeviceWarningListener处理
     * @param message mq消息
     * @param channel 处理mq消息的连接通道
     * @throws IOException IO类型异常，抛出以记录日志
     */
    @RabbitListener(queues = "rawDataQueues", ackMode="MANUAL", autoStartup = "false", id="readDeviceUploadMessage", concurrency="2")
    public void receiveMessage(Message message, Channel channel) throws IOException {

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try{
            if(null != message){
                String encoding = message.getMessageProperties().getContentEncoding();
                if(null == encoding){
                    encoding = "UTF-8";
                }
                String messageContent = new String(message.getBody(), encoding);
                log.info("Received Message {}:{} ", deliveryTag, messageContent);
                // 从mq取回的消息前后可能多 "  需要去除
                if(messageContent.startsWith("\"")){
                    messageContent = messageContent.substring(1, messageContent.length()-1);
                }
                if(messageContent.endsWith("\"")){
                    messageContent = messageContent.substring(0, messageContent.length()-1);
                }
                messageContent.replaceAll("\\\\","");
                // step1 解析出消息的设备

                JSONObject jsonObject = JSONObject.parseObject(messageContent);
                String uploadRegId = jsonObject.getString("deviceId");
                if(StringUtils.isNotBlank(uploadRegId)){
                    log.info("Received Message from uploadRegId:{} ", uploadRegId);
                    // step2 检查该设备对应的协议
                    TemplateDeviceUploadProtocol deviceUploadProtocol = getDeviceProtocol(uploadRegId);
                    if(deviceUploadProtocol == null){
                        log.warn("uploadRegId {} Message protocol is null or not find device with uploadRegId {}",uploadRegId,uploadRegId);
                        jsonObject.put("error", "Not find device with uploadRegId or Device Message Protocol is not defined in system");
                        rabbitTemplate.convertAndSend("rawDataErrorQueues", jsonObject);
                        channel.basicAck(deliveryTag,false);
                    }else{
                        String measurePointId = deviceUploadProtocol.getMeasurePointId();
                        Long tenantId = deviceUploadProtocol.getTenantId();
                        Long protocolId = deviceUploadProtocol.getId();
                        if(deviceUploadProtocol != null){
                            log.info("deviceId Message protocol is:{} ({}) ", deviceUploadProtocol.getProtocolName(), deviceUploadProtocol.getId());
                            //
                            String messageStr = jsonObject.getString("message");

                            LocalDateTime times = LocalDateTimeUtils.parseLocalDateTime(jsonObject.getString("timestamp"));
                            if(StringUtils.isNotBlank(messageStr)){
                                // 加载该消息对应的协议数据对象
                                Class<?> clazz = Class.forName("com.lnhuachuan.iot.arcel.arcm300.data.ARCM300StringData");
                                Object dataObj = clazz.getConstructor(String.class).newInstance(messageStr);
                                // 包装成系统内的消息对象 如果消息对象没有实现getCommandStr和getData 这两个方法将返回空
                                DeviceMessage deviceMessage = new DeviceMessageAdapter(dataObj);
                                // step3 取出该消息的指令码 判断是何类型消息
                                String commandStr = deviceMessage.getCommandStr();
                                log.debug("commandStr: {}",commandStr);
                                if(StringUtils.isNotBlank(commandStr)){
                                    TemplateDeviceUploadMessage messageTemplate = deviceUploadProtocol.getCommand(commandStr);
                                    if(messageTemplate != null){
                                        List<TemplateDeviceUploadDataConfig> dataTemplateList = messageTemplate.getTemplateDeviceUploadDataConfigList();
                                        if(dataTemplateList != null && !dataTemplateList.isEmpty()){
                                            // step4 根据消息类型，解析出其中的数据 存放到对应的数据表 包括时序数据库和关系数据库
                                            // 解析消息信息并存放到对应位置
//                                            List<Point> pointList = new ArrayList<>();
                                            List<BusinessDeviceUploadData> dataList = new ArrayList<>();
                                            String messageId = UUID.randomUUID().toString().replaceAll("-","");
                                            for(TemplateDeviceUploadDataConfig dataTemplate:dataTemplateList){
                                                Integer startBitIndex = dataTemplate.getStartBitIndex();
                                                Integer endBitIndex = dataTemplate.getEndBitIndex();

                                                BitDataTypeEnum bitDataType = BitDataTypeEnum.getByValue(dataTemplate.getBitDataType());
                                                if(bitDataType == null){
                                                    log.warn("bitDataType is null 可能是协议中bit_data_type配置错误，配置值是{}",dataTemplate.getBitDataType());
                                                }else{
                                                    // 取协议对应的消息体
                                                    byte[] byteData = deviceMessage.getData(startBitIndex, endBitIndex);
                                                    if(null != byteData){
                                                        String data = "";
                                                        switch (bitDataType){
                                                            case INT:
                                                            case DINT:
                                                                data = String.valueOf(HexUtils.bytesToInt(byteData));
                                                                break;
                                                            case FLOAT:
                                                            case REAL:
                                                            case LREAL:
                                                                log.debug("convert data to float:{}",HexUtils.byte2HexStr(byteData));
                                                                data = String.valueOf(HexUtils.floatByteArrToFloat(byteData, ByteOrder.LITTLE_ENDIAN));
                                                                break;
                                                            case STRING:
                                                            default:
                                                                data = HexUtils.asciiToString(byteData);
                                                        }
                                                        // 将解析出的数据存储数据库

                                                        BusinessDeviceUploadData deviceData = new BusinessDeviceUploadData();
                                                        deviceData.setDataTemplateId(dataTemplate.getId());
                                                        deviceData.setDataTimestamp(times);
                                                        deviceData.setDataValue(data);
                                                        deviceData.setDataType(dataTemplate.getMessageContentType());
                                                        deviceData.setDataUnit(dataTemplate.getDataUnit());
                                                        deviceData.setDeviceId(Long.valueOf(measurePointId));
                                                        deviceData.setMessageId(messageId); // 增加设备消息id  同一条设备消息上传的数据消息id相同
                                                        deviceData.setProtocolId(protocolId);
//                                                    deviceData.setTenantId(tenantId);
                                                        deviceData.setMessageTemplateId(messageTemplate.getId());
                                                        // 发布设备数据事件
                                                        applicationContext.publishEvent(new BusinessDeviceUploadDataEvent(deviceData));
//                                                    deviceUploadEventPublisher.publishCustomEvent(deviceData);
                                                        dataList.add(deviceData);
//                                                    Point point = Point
//                                                            .measurement("deviceUploadData")
//                                                            .addTag("measurePointId",measurePointId)
//                                                            .addTag("dataType",dataTemplate.getMessageContentType())
//                                                            .addTag("dataUnit",dataTemplate.getDataUnit())
//                                                            .addField("dataValue",data)
//                                                            .time(Instant.now(), WritePrecision.NS);
//                                                    pointList.add(point);

                                                    }else{
                                                        log.warn("measurePointId {}  Protocol not support getData Method, MessageContent is {}",measurePointId,messageContent);
                                                    }
                                                }
                                            }
                                            // 输出到influxDB 此方法移交给spring event事件处理
//                                            writeToInfluxDB(pointList);

                                            // 输出到mysql 此方法为同步
                                            businessDeviceUploadDataService.saveBatch(tenantId,dataList);
                                            log.info("Message has saved from measurePointId:{} ", measurePointId);
                                            // 返回消息确认
                                            channel.basicAck(deliveryTag,false);
                                            // 处理数据报警规则
                                            // 这里只需要处理系统内定义的数据报警规则 设备的不需要处理
                                            // 此处改为由events处理
//                                            List<BusinessDeviceUploadWarningConfig> list = warningConfigService.getDeviceWarningConfig(measurePointId);
//                                            for(BusinessDeviceUploadWarningConfig config : list) {
//                                                for(BusinessDeviceUploadData payload:dataList){
//                                                    checkWarningType(config,payload);
//                                                }
//                                            }
                                        }else{
                                            // 未配置消息模板 ==> 没有数据需要读取
                                            log.warn("measurePointId {} MessageContent command dataTemplateList null, MessageContent is {}",measurePointId,messageContent);
                                            channel.basicAck(deliveryTag,false);
                                        }
                                    }else{
                                        // 未配置消息模板 ==> 不被系统认可的指令类型
                                        log.warn("measurePointId {} MessageContent command is null, MessageContent is {}",measurePointId,messageContent);
                                        jsonObject.put("error ", "MessageContent command is not defined in MessageTemplate");
                                        writeErrorToMq(channel, jsonObject, deliveryTag);
                                    }
                                    // 这里处理设备报警的信息 报警内容由协议定义
                                    List<BusinessDeviceWarnRecord> deviceMessageEvents = deviceMessage.getEvents();
                                    if(null!=deviceMessageEvents && !deviceMessageEvents.isEmpty()){
                                        for(BusinessDeviceWarnRecord event : deviceMessageEvents){
                                            event.setMeasureId(Long.valueOf(measurePointId));
                                            event.setDeviceData(messageContent);
                                            event.setDeviceTime(times);
                                            // 下面两条属性由协议内置适配器处理
//                                        event.setWarningContent(s);
//                                        event.setWarningCheckType("DEVICE_UPLOAD");
                                            event.setWarningLevel("0"); // 设备报警
                                            applicationContext.publishEvent(new BusinessDeviceWarnRecordEvent(event));
//                                            deviceWarningEventPublisher.publishCustomEvent(event);
                                        }
                                    }

                                }else{
                                    // 消息体为空
                                    log.warn("measurePointId {} MessageContent command is null or Protocol not support getCommandStr Method, MessageContent is {}",measurePointId,messageContent);
                                    jsonObject.put("error", "MessageContent command is null or Protocol not support getCommandStr Method");
                                    writeErrorToMq(channel, jsonObject, deliveryTag);
                                }
                            }else{
                                // 消息体为空
                                log.warn("measurePointId {} MessageContent not contain attribute[message], MessageContent is {}",measurePointId,messageContent);
                                jsonObject.put("error", "MessageContent not contain attribute[message]");
                                writeErrorToMq(channel, jsonObject, deliveryTag);
                            }

                        }else{
                            log.warn("measurePointId {} Message protocol is null",measurePointId);
                            jsonObject.put("error", "Device Message Protocol is not defined in system");
                            writeErrorToMq(channel, jsonObject, deliveryTag);
                        }
                    }

                }else{
                    log.warn("uploadRegId is null");
                    jsonObject.put("error", "uploadRegId is null");
                    writeErrorToMq(channel, jsonObject, deliveryTag);
                }
            }else{
                // 正常情况下message不应该为null  这段纯纯是为了处理某次出现的message异常
                log.warn("message is null");
                rabbitTemplate.convertAndSend("rawDataErrorQueues", message);
                channel.basicAck(deliveryTag, false);
            }

        } catch (Exception e){
            // 出现了预期之外的异常 不消费此消息
            rabbitTemplate.convertAndSend("rawDataErrorQueues", message);
            channel.basicAck(deliveryTag, false);
            e.printStackTrace();
        } finally {
            log.debug("message ack {}", deliveryTag);
        }

    }

    /**
     * 将错误消息写入mq异常消息队列 同时发送ack消费确认
     *
     * @param channel
     * @param jsonObject
     * @param deliveryTag
     * @throws IOException
     */
    private void writeErrorToMq(Channel channel, JSONObject jsonObject, long deliveryTag) throws IOException {
        rabbitTemplate.convertAndSend("rawDataErrorQueues", jsonObject);
        channel.basicAck(deliveryTag,false);
    }



    /**
     * 根据regId取设备所用的协议信息
     * @param uploadRegId 设备注册id mq中为regId
     * @return TemplateDeviceUploadProtocol 设备id对应的协议的完整信息
     * 缓存中有记录则返回缓存中内容，如不存在将去数据库查询并返回，同时更新缓存记录
     */
    public TemplateDeviceUploadProtocol getDeviceProtocol(String uploadRegId) {
        String cacheKey = RedisKeyConstants.CACHED_PROTOCOL_HEAD + uploadRegId;
        log.debug("cacheKey： {}", cacheKey);
        TemplateDeviceUploadProtocol deviceUploadProtocol = null;
        if(redisTemplate.hasKey(cacheKey)){
            log.debug("cacheKey 命中");
            deviceUploadProtocol= (TemplateDeviceUploadProtocol) redisTemplate.opsForValue().get(cacheKey);
        }else{
            log.debug("cacheKey 未命中");
            deviceUploadProtocol = templateDeviceUploadProtocolService.getMeasurePointProtocol(uploadRegId);
            if(null!=deviceUploadProtocol){
                log.debug("deviceUploadProtocol {}", deviceUploadProtocol);
                redisTemplate.opsForValue().set(cacheKey, deviceUploadProtocol);
            }

        }
        return deviceUploadProtocol;
    }





}
